package org.example.process;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.flink.pojo.WaterSensor;
import org.example.function.WaterSensorFunction;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/2/23 22:01
 */
public class TopNProcessAllWindowFun {

    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        environment.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy1 =
                WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(
                                (TimestampAssignerSupplier<WaterSensor>) context ->
                                        (element, recordTimestamp) -> element.getVc() * 1000L
                        );

        SingleOutputStreamOperator<String> streamOperator =
                environment.socketTextStream("127.0.0.1", 7777)
                        .map(new WaterSensorFunction())
                        .assignTimestampsAndWatermarks(watermarkStrategy1)
                        .windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                        .process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>() {
                            @Override
                            public void process(ProcessAllWindowFunction<WaterSensor, String, TimeWindow>.Context context,
                                                Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                                Map<Integer, Integer> vcCountMap = new HashMap<>();

                                for (WaterSensor element : elements) {
                                    Integer vc = element.getVc();
                                    if (vcCountMap.containsKey(vc)) {
                                        vcCountMap.put(vc, vcCountMap.get(vc) + 1);
                                    } else {
                                        vcCountMap.put(vc, 1);
                                    }
                                }

                                List<Tuple2<Integer, Integer>> datas = new ArrayList<>();
                                for (Map.Entry<Integer, Integer> entry : vcCountMap.entrySet()) {
                                    Integer vc = entry.getKey();
                                    datas.add(Tuple2.of(vc, vcCountMap.get(vc)));
                                }
                                datas.sort((o1, o2) -> {
                                    // 降序， 后 减 前
                                    return o2.f1 - o1.f1;
                                });

                                StringBuilder outStr = new StringBuilder();
                                outStr.append("窗口开始时间： ")
                                        .append(DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss.SSS"))
                                        .append("\n");
                                for (int i = 0; i < Math.min(2, datas.size()); i++) {
                                    Tuple2<Integer, Integer> vcCount = datas.get(i);
                                    outStr.append("Top").append(i + 1).append("\n");
                                    outStr.append("vc=").append(vcCount.f0).append(",");
                                    outStr.append("count=").append(vcCount.f1).append("\n");
                                }
                                outStr.append("窗口结束时间=")
                                        .append(DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss.SSS"))
                                        .append("\n");
                                outStr.append("================================\n");
                                out.collect(outStr.toString());

                            }
                        });

        streamOperator.print();

        environment.execute();

    }

}
